package com.omenkk.juc.execute;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class ThreadPool {
    //任务队列
    private BlockingQueue<Runnable> taskQueue;

    //核心线程数
    private int coreSize;

    //线程集合
    private Set<Worker> workers=new HashSet<>();
    //超时时间
    private long timeout;

    private TimeUnit timeUnit;

    //拒绝策略
    private  RejectPolicy rejectPolicy;

    public ThreadPool(BlockingQueue<Runnable> taskQueue, int coreSize, long timeout, TimeUnit timeUnit, RejectPolicy rejectPolicy) {
        this.taskQueue = taskQueue;
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.rejectPolicy = rejectPolicy;
    }

    public void execute(Runnable task){
        synchronized (workers){
            //如果线程数没有超过线程数量
            if(workers.size()<coreSize){
                Worker worker = new Worker(task);
                //添加任务线程到线程池
                workers.add(worker);
                //启动线程
                worker.start();
            }else{
//添加到队列(携带拒绝策略)
                taskQueue.tryPut(this.rejectPolicy,task);
            }
        }
    }
    //任务线程
    class Worker extends Thread{
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            //当任务不为空，处理任务
            //当任务为空，从阻塞队列获取任务
            while(task!=null||(task=taskQueue.poll(timeout,timeUnit))!=null){
                try{
                    //在任务线程中执行任务
                    log.debug("正在执行...{}",task);
                    task.run();
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    task=null;
                }
                synchronized (workers){
                    //销毁线程
                    log.debug("工作线程被移除:{}",this);
                    workers.remove(this);
                }
            }
        }
    }
}
@Slf4j
enum RejectPolicy{

    /**
     *死等
     */
//    WAIT{
//        @Override
//        public <T> void tryPut(BlockingQueue<T> tBlockingQueue, T task) {
//            tBlockingQueue.put(task);
//        }
//    },



    /**
     * 超时等待
     */
    TIMED_WAIT{
        @Override
        public <T> void tryPut(BlockingQueue<T> tBlockingQueue,T task) {
boolean success=tBlockingQueue.put(task,tBlockingQueue.getTimeout(),tBlockingQueue.getTimeUnit());
if(!success){
    throw new RuntimeException("Queue full after timed wait");
}
        }
    },
    /**
     * 调用者自己执行任务
     */
    CALLER_RUNS{
        @Override
        public <T> void tryPut(BlockingQueue<T> tBlockingQueue,T task) {
if(task instanceof Runnable){
   ((Runnable) task).run();
}
        }
        /**
         *
         */
    },
    /**
     * 抛出异常
     */
    ABORT{
        @Override
        public <T> void tryPut(BlockingQueue<T> tBlockingQueue, T task) {
            throw new RuntimeException("Task rejected: " + task);
        }
    },
    /**
     * 放弃处理
     */
    DISCARD{
        @Override
        public <T> void tryPut(BlockingQueue<T> tBlockingQueue, T task) {
            log.debug("Task discarded: {}",task);
        }
    }








    ;

    public abstract  <T> void tryPut(BlockingQueue<T> tBlockingQueue,T task);


}
//任务线程

//阻塞任务队列
@Getter
class BlockingQueue<T>{

    private long timeout;

    private TimeUnit timeUnit;
//双端队列
    private Deque<T> queue;
//锁
    private ReentrantLock lock;

    //生产者条件变量
    private Condition fullWaitSet;
    //消费者条件变量
    private Condition emptyWaitSet;
    //队列容量
    int capacity;

    public BlockingQueue(long timeout, TimeUnit timeUnit, int capacity) {
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.queue =new ArrayDeque<>(capacity);
        this.lock =new ReentrantLock();
        this.fullWaitSet = this.lock.newCondition();
        this.emptyWaitSet =this.lock.newCondition();
        this.capacity = capacity;
    }

    //带超时的取出任务
    public T poll(long timeout, TimeUnit timeUnit){
        lock.lock();
        try{
            //统一单位
            long nanos=  timeUnit.toNanos(timeout);
            while(queue.isEmpty()){
                //进入等待
                try{
                    //超时
                    if(nanos<=0){
                        return null;
                    }
                    //返回剩余时间
                  nanos= emptyWaitSet.awaitNanos(nanos);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
//执行获取操作
            T task = queue.removeFirst();
            fullWaitSet.signal();
            return task;
        }finally {
            lock.unlock();
        }
    }
    //取出任务
    public T take(){
lock.lock();
try{
    while(queue.isEmpty()){
        //进入等待
        try{
            emptyWaitSet.await();
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
//执行获取操作
    T task = queue.removeFirst();
    fullWaitSet.signalAll();
    return task;
}finally {
    lock.unlock();
}

    }
    //放入任务
    public void put(T task){
        lock.lock();
        try{
            while(queue.size()==capacity){
                try{
                    fullWaitSet.await();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
                //当队列满了
                //等待，直到队列不满
            }
            //执行操作
            queue.addLast(task);
            //唤醒
            emptyWaitSet.signalAll();
        }finally {
            lock.unlock();
        }

    }
    //放入任务
    public boolean put(T task, long timeout, TimeUnit timeUnit){
        lock.lock();
        try{
            long nanos=timeUnit.toNanos(timeout);
            while(queue.size()==capacity){
                if(nanos<=0) return false;
                try{
                    nanos=fullWaitSet.awaitNanos(nanos);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
                //当队列满了
                //等待，直到队列不满
            }
            //执行操作
            queue.addLast(task);
            //唤醒
            emptyWaitSet.signalAll();
        }finally {
            lock.unlock();
        }
        return true;
    }
    public int size(){
        lock.lock();
        try{
            return queue.size();
        }finally {
            lock.unlock();
        }
    }


    public void tryPut(RejectPolicy rejectPolicy, T task) {
        lock.lock();
        //如果队列满了，交给拒绝策略执行
        try{
            if(queue.size()==capacity){
rejectPolicy.tryPut(this,task);
            }else{
                //队列没有满，添加到队列
                queue.addLast(task);
                //提醒消费者
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }

    }
}